package ex.datastream.functions.dataStream;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.IterativeStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.api.common.functions.FilterFunction;

/**
 * https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/datastream/operators/overview/#iterate

 * ds1.map(+1) 得到:
 *  Datastream  initialStream =  2 ,3 ,4,  5 ,6, 7 ,8 ,9
 *
 * iteration =  initialStream.iterate()
 *
 * iteration.map(+40)  =  42 43 ,44,  45 ,46, 47 , 48 ,49
 *
 * feedback = iteration(<43)  = 42
 *
 * output =  iteration(>=43)  = 43 ,44,  45 ,46, 47 , 48 ,49
 *
 * 重点:
 *     iteration.closeWith(feedback);
 *     此时iteration尾部增加了原始  42
 *
 *     然后进行下一轮:
 *
 *                 iteration.map(+40)  = 42+40 = 82
 *                 feedback = iteration(<43)  此时没有元素了.
 *                 output = 82
 *
 *      因为feedback第二轮无元素追加, iteration.next()没有元素.所以程序处于等待状态,不再有运行
 */
public class IterateFunc {
	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		DataStream<Tuple2<String,Integer> > ds1 = env
				.fromElements(
						new Tuple2<String,Integer>("zy",1),
						new Tuple2<String,Integer>("zy",2),
						new Tuple2<String,Integer>("zy",3),
						new Tuple2<String,Integer>("ll",4),
						new Tuple2<String,Integer>("ll",5),
						new Tuple2<String,Integer>("ll",6),
						new Tuple2<String,Integer>("zy",7),
						new Tuple2<String,Integer>("ll",8)
				);


		DataStream<Long> initialStream = ds1.map(
				new MapFunction<Tuple2<String,Integer>, Long>() {
					@Override
					public Long map(Tuple2<String,Integer> value) throws Exception {
						return    1L + value.f1;
					}
				});

//		initialStream.print();
		IterativeStream<Long> iteration = initialStream.iterate();
		DataStream<Long> iterationBody= iteration.map(value -> value +40);

		DataStream<Long> feedback = iterationBody.filter(new FilterFunction<Long>(){
			@Override
			public boolean filter(Long value) throws Exception {
				return value < 43;
			}
		});
		iteration.closeWith(feedback);

		feedback.print();
		DataStream<Long> output = iterationBody.filter(new FilterFunction<Long>(){
			@Override
			public boolean filter(Long value) throws Exception {
				return value >= 43;
			}
		});

		output.print();
		System.out.println("                 \n\n  ");


		env.execute();
	}


}
